package com.finfosoft.mqtt;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import com.alibaba.fastjson.JSON;
import com.jfinal.kit.Prop;
import com.jfinal.kit.PropKit;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Record;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 发布消息的回调类
 * 
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。 在回调中，将它用来标识已经启动了该回调的哪个实例。
 * 必须在回调类中实现三个方法：
 * 
 * public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 * 
 * public void connectionLost(Throwable cause)在断开连接时调用。
 * 
 * public void deliveryComplete(MqttDeliveryToken token)) 接收到已经发布的 QoS 1 或 QoS 2
 * 消息的传递令牌时调用。 由 MqttClient.connect 激活此回调。
 * 
 */
public class PushCallback implements MqttCallback {

	private final static Logger log = LoggerFactory.getLogger(PushCallback.class);
	
	Prop prop = PropKit.use("config.properties");

	public void connectionLost(Throwable cause) {
		//	log.info("连接断开，可以做重连");
		if(prop.get("mqtt.clientid")!=null){
			try {
				Thread.sleep(30000);
				new Client().start();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public void deliveryComplete(IMqttDeliveryToken token) {
		//System.out.println("deliveryComplete---------" + token.isComplete());
	}

	public void messageArrived(String topic, MqttMessage message) throws Exception {
		// subscribe后得到的消息会执行到这里面
		
		log.info("接收消息主题 : " + topic + ",接收消息内容 : " + new String(message.getPayload()));
		List<Map<String, Object>> datas = (List<Map<String, Object>>) JSON.parse(new String(message.getPayload()));
		String sql = null;
		log.info("保存前的数量："+Db.find("select * from "+prop.get("mqtt.oper_dbname")).size());
		List<Record> configs = new ArrayList<>();
		for (Map<String, Object> data : datas) {
			int dataId = (int) data.get("data_id");
			sql = "delete from " + prop.get("mqtt.oper_dbname") + " where data_id=" + dataId;
			Db.update(sql);
			Record record = new Record();
			record.setColumns(data);
			configs.add(record);
		}
		int num = prop.get("mqtt.oper_dbcolumn").split(",").length;
		String columnNum = "?";
		for (int i = 0; i < num-1; i++) {
			columnNum += ",?";
		}
		sql = "insert into " + prop.get("mqtt.oper_dbname")+" ( "+prop.get("mqtt.oper_dbcolumn")+") values("+columnNum+")";
		Db.batch(sql,prop.get("mqtt.oper_dbcolumn"), configs, 50);
		log.info("保存后的数量："+Db.find("select * from "+prop.get("mqtt.oper_dbname")).size());
	}

}
